package com.ec.rabbitmq.reciver;

import cn.hutool.core.date.DateTime;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;


/**
 * basicRecover：是路由不成功的消息可以使用recovery重新发送到队列中。
 * basicReject：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是
 * 否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能
 * 批量拒绝消息，为解决批量拒绝消息才有了basicNack。
 * basicNack：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple
 * 参数为true，服务器会拒绝指定了delivery_tag的所有未
 * 确认的消息(tag是一个64位的long值，最大值是9223372036854775807)。
 *
 * @author Eric Lee
 * @since 2022-07-14 21:24
 */
@Slf4j
@Component
public class RabbitReceiver {

    /**
     * @param data
     * @param consumerTag
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "${rabbit.queue.direct-queue:ec-direct-queue}")
    public void directHandler(@Payload String data, @Header(AmqpHeaders.CONSUMER_TAG) String consumerTag,
                              @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        try {
            if (data.contains("傻逼")) {
                throw new NullPointerException("模拟在实际业务中异常~~");
            } else {
                // 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息（成功消费，消息从队列中删除）
                channel.basicAck(deliveryTag, false);
            }
        } catch (IOException e) {
            //重新入列
            channel.basicNack(deliveryTag, false, true);
        } catch (Exception e) {
            //业务异常进入死信队列
            channel.basicNack(deliveryTag, false, false);
        }
    }

    /**
     * topic
     *
     * @param data
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = "${rabbit.queue.topic-queue:ec-topic-queue}")
    public void topicHandler(@Payload String data, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        try {
            log.info("收到的topic消息：{}", data);
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(queues = "${rabbit.queue.fanout-queue:ec-fanout-queue}")
    public void fanoutHandler(@Payload String data, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        try {
            log.info("收到的fanout消息：{}", data);
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            channel.basicNack(deliveryTag, false, false);
        }
    }

    /**
     * 死信队列
     *
     * @param message
     * @param data
     * @param channel
     */
    @RabbitListener(queues = "${rabbit.queue.dead-letter-queue:ec-dead-letter-queue}")
    public void deadLetterHandler(Message message, @Payload String data, Channel channel) {
        String fromExchange = message.getMessageProperties().getHeader("x-first-death-exchange").toString();
        String fromQueue = message.getMessageProperties().getHeader("x-first-death-queue").toString();
        String reason = message.getMessageProperties().getHeader("x-first-death-reason").toString();
        JSONObject xDeath = JSON.parseObject(JSON.toJSONString(message)).getJSONObject("messageProperties")
                .getJSONObject("headers").getJSONArray("x-death").getJSONObject(0);
        DateTime time = DateTime.of(xDeath.getLong("time"));
        String formRoutingKeys = xDeath.getString("routing-keys");
        log.info("在{}收到的死信队列消息：{},原因={},fromExchange={},fromQueue={},fromRoutingKey={}",
                time, data, reason, fromExchange, fromQueue, formRoutingKeys);
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
